package com.yanfan.mq.redischannel.consumer;

import com.yanfan.common.constant.YanfanConstant;
import com.yanfan.common.core.mq.message.DeviceDownMessage;
import com.yanfan.common.core.mq.message.ModbusPollMsg;
import com.yanfan.common.core.mq.message.PropRead;
import com.yanfan.common.core.protocol.Message;
import com.yanfan.common.core.redis.RedisCache;
import com.yanfan.common.core.redis.RedisKeyBuilder;
import com.yanfan.common.enums.ServerType;
import com.yanfan.common.enums.TopicType;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.common.utils.gateway.mq.TopicsUtils;
import com.yanfan.common.utils.gateway.protocol.ByteUtils;
import com.yanfan.iot.service.IFunctionLogService;
import com.yanfan.mq.service.impl.MessageManager;
import com.yanfan.mqttclient.PubMqttClient;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 平台定时批量获取设备属性(或单个获取)
 *
 * @author bill
 */
@Slf4j
@Component
public class DevicePropFetchConsumer {

    //锁集合
    private final ConcurrentHashMap<String, Lock> taskLocks = new ConcurrentHashMap<>();
    @Autowired
    private PubMqttClient pubMqttClient;
    @Autowired
    private RedisCache redisCache;
    @Resource
    private MessageManager messageManager;
    @Resource
    private TopicsUtils topicsUtils;
    @Resource
    private IFunctionLogService functionLogService;

    @Async(YanfanConstant.TASK.DEVICE_FETCH_PROP_TASK)
    public void consume(ModbusPollMsg task) {
        if (Objects.nonNull(task.getData())) {
            if (task.getData() instanceof DeviceDownMessage) {
                execute((DeviceDownMessage) task.getData());
            }
        } else {
            execute1(task);
        }
    }


    private void execute1(ModbusPollMsg task) {
        String serialNumber = task.getSerialNumber();
        //获取一个线程锁
        Lock lock = taskLocks.computeIfAbsent(serialNumber, k -> new ReentrantLock());
        //阻塞直到获取到锁
        lock.lock();
        try {
            Long productId = task.getProductId();
            List<String> commandList = task.getCommandList();
            ServerType serverType = ServerType.explain(task.getTransport());
            String topic = topicsUtils.buildTopic(productId, serialNumber, TopicType.FUNCTION_GET);
            for (String command : commandList) {
                String cacheKey = RedisKeyBuilder.buildModbusRuntimeCacheKey(serialNumber);
                redisCache.zSetAdd(cacheKey, command, DateUtils.getTimestampSeconds());
                switch (serverType) {
                    //通过mqtt内部客户端 下发指令
                    case MQTT:
                        publish(topic, ByteBufUtil.decodeHexDump(command));
                        log.info("=>MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread
                                .currentThread()
                                .getName(), command, topic);
                        break;
                    //  下发TCP客户端
                    case TCP:
                        Message msg = new Message();
                        msg.setClientId(serialNumber);
                        msg.setPayload(Unpooled.wrappedBuffer(ByteBufUtil.decodeHexDump(command)));
                        messageManager.requestR(serialNumber, msg, Message.class);
                        log.info("=>TCP-线程=[{}],轮询指令:[{}]", Thread.currentThread().getName(), command);
                        break;
                }
                //指令间隔时间
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            log.error("线程错误e", e);
        } finally {
            lock.unlock();
        }
    }

    public void publish(String topic, byte[] pushMessage) {
        redisCache.incr2(YanfanConstant.REDIS.MESSAGE_SEND_TOTAL, -1L);
        redisCache.incr2(YanfanConstant.REDIS.MESSAGE_SEND_TODAY, 60 * 60 * 24);
        pubMqttClient.publish(pushMessage, topic, false, 0);
    }


    /**
     * 平台自动发送指令函数
     *
     * @param message
     */
    private void execute(DeviceDownMessage message) {
        try {
            for (PropRead read : message.getValues()) {
                //缓存值
                String cacheKey = RedisKeyBuilder.buildPropReadCacheKey(message.getSubCode());
                redisCache.setCacheObject(cacheKey, read, 300000, TimeUnit.MILLISECONDS);
                switch (message.getServerType()){
                    //通过mqtt内部客户端 下发指令
                    case MQTT:
                        publish(message.getTopic(), ByteUtils.hexToByte(read.getData()));
                        log.info("=>MQTT-线程=[{}],轮询指令:[{}],主题:[{}]，设备编号：{}", Thread
                                .currentThread()
                                .getName(), read.getData(), message.getTopic(), message.getSerialNumber());
                        break;
                    //  下发TCP客户端
                    case TCP:
                        Message msg = new Message();
                        msg.setClientId(message.getSerialNumber());
                        msg.setPayload(Unpooled.wrappedBuffer(ByteBufUtil.decodeHexDump(read.getData())));
                        messageManager.requestR(message.getSerialNumber(), msg, Message.class);
                        log.info("=>TCP-线程=[{}],轮询指令:[{}]，设备编号：{}", Thread
                                .currentThread()
                                .getName(), read.getData(), message.getSerialNumber());
                        break;
                }
                Thread.sleep(1500);
            }
        } catch (Exception e) {
            log.error("线程错误e",e);
        }
    }
}
